package com.itheima.behavior.config;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.errors.LogAndContinueExceptionHandler;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.annotation.EnableKafkaStreams;
import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaStreamsConfiguration;
import org.springframework.kafka.core.*;

import java.util.HashMap;
import java.util.Map;

/**
 * Kafka 配置类
 * 配置 Kafka Producer、Consumer 和 Streams
 */
@Configuration
@EnableKafka
@EnableKafkaStreams
@Slf4j
public class KafkaConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${spring.kafka.consumer.group-id}")
    private String groupId;

    @Value("${spring.kafka.streams.application-id}")
    private String streamsApplicationId;

    /**
     * Kafka Producer 配置
     */
    @Bean
    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        config.put(ProducerConfig.ACKS_CONFIG, "1");
        config.put(ProducerConfig.RETRIES_CONFIG, 3);
        config.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        config.put(ProducerConfig.LINGER_MS_CONFIG, 10);
        config.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        // 压缩配置 - lz4 压缩率和性能平衡最好
        config.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");
        // 请求超时时间
        config.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
        // 最大请求大小 1MB
        config.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 1048576);
        return new DefaultKafkaProducerFactory<>(config);
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

    /**
     * Kafka Consumer 配置
     */
    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        config.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        config.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);
        return new DefaultKafkaConsumerFactory<>(config);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = 
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);
        return factory;
    }

    /**
     * Kafka Streams 配置
     */
    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
    public KafkaStreamsConfiguration kStreamsConfig() {
        Map<String, Object> props = new HashMap<>();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, streamsApplicationId);
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 3);
        props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000);
        
        // 异常处理策略：记录日志并继续
        props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
                LogAndContinueExceptionHandler.class);
        
        // 状态存储目录
        props.put(StreamsConfig.STATE_DIR_CONFIG, System.getProperty("java.io.tmpdir") + "/kafka-streams");
        
        // 优化配置
        props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 10 * 1024 * 1024L); // 10MB 缓存
        props.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 1); // 单节点副本数为1
        
        log.info("========== Kafka Streams 配置初始化 ==========");
        log.info("Application ID: {}", streamsApplicationId);
        log.info("Bootstrap Servers: {}", bootstrapServers);
        log.info("Stream Threads: 3");
        log.info("==========================================");
        
        return new KafkaStreamsConfiguration(props);
    }

    /**
     * Kafka Topic 常量
     */
    public static class Topics {
        // 用户行为原始事件流
        public static final String USER_BEHAVIOR = "user-behavior";
        
        // 图书相关行为事件
        public static final String BOOK_BEHAVIOR = "book-behavior";
        
        // 排行榜更新事件
        public static final String RANKING_UPDATE = "ranking-update";
    }

    /**
     * Kafka Key 常量
     */
    public static class Keys {
        // ========== 实时排行榜（Kafka Streams 计算）==========
        // 实时热度排行榜（近期热度，适合发现新热点）
        public static final String BOOK_REALTIME_HOT_RANKING = "ranking:book:realtime:hot";
        
        // 实时阅读排行榜
        public static final String BOOK_REALTIME_READ_RANKING = "ranking:book:realtime:read";
        
        // 实时收藏排行榜
        public static final String BOOK_REALTIME_COLLECT_RANKING = "ranking:book:realtime:collect";
        
        // 实时点赞排行榜
        public static final String BOOK_REALTIME_LIKE_RANKING = "ranking:book:realtime:like";
        
        // ========== 综合排行榜（定时任务计算，带时间衰减）==========
        // 综合热度排行榜（历史累积+时间衰减，适合发现经典热门）
        public static final String BOOK_HOT_RANKING = "ranking:book:hot";
        
        // 图书阅读排行榜（总阅读人数）
        public static final String BOOK_READ_RANKING = "ranking:book:read";
        
        // 图书收藏排行榜（总收藏数）
        public static final String BOOK_COLLECT_RANKING = "ranking:book:collect";
        
        // 图书点赞排行榜（总点赞数）
        public static final String BOOK_LIKE_RANKING = "ranking:book:like";
        
        // ========== 用户排行榜 ==========
        // 用户阅读时长排行榜
        public static final String USER_READING_TIME_RANKING = "ranking:user:reading_time";
        
        // 用户活跃度排行榜
        public static final String USER_ACTIVE_RANKING = "ranking:user:active";
        
        // ========== 时间段排行榜 ==========
        // 今日热度榜
        public static final String BOOK_TODAY_HOT_RANKING = "ranking:book:today:hot";
        
        // 本周热度榜
        public static final String BOOK_WEEK_HOT_RANKING = "ranking:book:week:hot";
        
        // 本月热度榜
        public static final String BOOK_MONTH_HOT_RANKING = "ranking:book:month:hot";
        
        // ========== 笔记排行榜 ==========
        // 实时热门笔记排行榜（Kafka Streams 计算）
        public static final String NOTE_REALTIME_HOT_RANKING = "ranking:note:realtime:hot";
        
        // 综合热门笔记排行榜（定时任务计算）
        public static final String NOTE_HOT_RANKING = "ranking:note:hot";
        
        // 笔记点赞排行榜
        public static final String NOTE_LIKE_RANKING = "ranking:note:like";
        
        // 笔记收藏排行榜
        public static final String NOTE_COLLECT_RANKING = "ranking:note:collect";
        
        // ========== 书评排行榜 ==========
        // 实时热门书评排行榜（Kafka Streams 计算）
        public static final String REVIEW_REALTIME_HOT_RANKING = "ranking:review:realtime:hot";
        
        // 综合热门书评排行榜（定时任务计算）
        public static final String REVIEW_HOT_RANKING = "ranking:review:hot";
        
        // 书评点赞排行榜
        public static final String REVIEW_LIKE_RANKING = "ranking:review:like";
        
        // 书评收藏排行榜
        public static final String REVIEW_COLLECT_RANKING = "ranking:review:collect";
    }
}

